package com.clown.hotItemsAnalysis


import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Slide}
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row


object HotItemsWithSql {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 定义事件时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val resource = getClass.getResource("/UserBehavior.csv")
    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

    // 从文件中读取数据，并转换成样例类，并提取时间戳生成watermark
    val dataStream: DataStream[UserBehavior] = inputStream
      .map(data => {
        val arr = data.split(",")
        UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toLong, arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    // 定义表执行环境
    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv = StreamTableEnvironment.create(env, settings)

    // 基于dataStream创建table
    val dataTable = tableEnv.fromDataStream(dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)

    // 1. table api 进行开窗聚合统计
    val aggTable = dataTable
      .filter('behavior === "pv")
      .window(Slide over 1.hours every 5.minutes on 'ts as 'sw)
      .groupBy('itemId, 'sw)
      .select('itemId, 'sw.`end` as 'windowEnd, 'itemId.count as 'cnt)

    // 使用sql去实现TopN的选取
    tableEnv.createTemporaryView("aggTable", aggTable, 'itemId, 'windowEnd, 'cnt)
    val resultTable = tableEnv.sqlQuery(
      """
        |select
        | *
        |from (
        | select
        |   *,
        |   row_number()
        |     over (partition by windowEnd order by cnt desc)
        |     as row_num
        | from aggTable
        |)
        |where row_num<=5
        |""".stripMargin)

    // 纯sql实现
    tableEnv.createTemporaryView("datatable", dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)
    val resultSqlTable = tableEnv.sqlQuery(
      """
        |select
        | *
        |from (
        | select
        |   *,
        |   row_number()
        |     over (partition by windowEnd order by cnt desc)
        |     as row_num
        | from (
        |   select
        |     itemId,
        |     hop_end(ts,interval '5' minute, interval '1' hour) as windowEnd,
        |     count(itemId) as cnt
        |   from datatable
        |   where behavior='pv'
        |   group by
        |     itemId,
        |     hop(ts,interval '5' minute, interval '1' hour)
        | )
        |)
        |where row_num<=5
        |""".stripMargin)

    //    resultTable.toRetractStream[Row].print("agg")
    resultSqlTable.toRetractStream[Row].print("agg")

    env.execute("hot items with sql")
  }
}
